/**
 * Copyright 2019 asiainfo Inc.
 **/
package com.myz.rabbitmq.rpc;

import com.myz.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * Rpc客户端
 * 1. 配置连接工厂
 * 2. 建立TCP连接
 * 3. 在TCP连接的基础上创建通道
 * 4. 定义临时队列replyQueueName，声明唯一标志本次请求corrId，并将replyQueueName和corrId配置要发送的消息队列中
 * 5. 使用默认的交换机发送消息到队列rpc_queue中
 * 6. 使用阻塞队列BlockingQueue阻塞当前进程
 * 7. 收到请求后，将请求放入BlockingQueue中，主线程被唤醒，打印返回内容
 *
 * @author maoyz0621 on 19-1-15
 * @version: v1.0
 */
public class RpcClient {

    private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);
    private static final String RPC_QUEUE_NAME = "rpc_queue";

    public static void execute(String message) throws Exception {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = null;

        try {
            // 在TCP连接的基础上创建通道
            channel = connection.createChannel();
            // 定义临时队列，并返回生成的队列名称
            String replyQueueName = channel.queueDeclare().getQueue();
            logger.debug("replyQueueName = {}", replyQueueName);

            // 唯一标志本次请求
            String corrId = UUID.randomUUID().toString();
            // 生成发送消息的属性
            AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .correlationId(corrId) // 唯一标志本次请求
                    .replyTo(replyQueueName) // 设置回调队列
                    .build();

            // 发送消息，发送到默认交换机
            channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes("UTF-8"));

            logger.debug("[RpcClient] Requesting : " + message);

            // 阻塞队列，用于存储回调结果
            final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);

            // 定义消息的回退方法
            channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {

                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    if (properties.getCorrelationId().equals(corrId)) {
                        response.offer(new String(body, "UTF-8"));
                    }
                }
            });

            // 获取回调的结果
            String result = response.take();
            logger.debug(" [RpcClient] Result:'" + result + "'");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                // 空值判断，为了代码简洁略
                channel.close();
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
